package com.atbeijing.D09;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * 处理时间,开窗
 * 求每个sensor每小时的平均温度
 */
public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //流数据
        SingleOutputStreamOperator<SensorReading> stream = env.addSource(new SensorSource()).filter(r -> r.id.equals("sensor_1"));

        //配置表环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();

        //获取表环境
        StreamTableEnvironment TEnv = StreamTableEnvironment.create(env, settings);

        //table api
        Table table = TEnv.fromDataStream(
                stream,
                $("id"),
                $("temperature").as("temp"),
                $("timestamp").as("ts"),
                $("pt").proctime()//添加一个字段,作为处理时间
        );
        //查询
        Table selectResult = table
                .window(Tumble.over(lit(5).seconds()).on($("pt")).as("w"))//开一个5s的滚动窗口
                .groupBy($("id"), $("w"))//相当于一个不断增加变化的表,每隔5秒组成一个小表,小表内根据id分组
                .select($("id"), $("temp").avg(), $("w").start());
        //获取结果
        TEnv.toRetractStream(selectResult, Row.class).print();

        //sql api
        TEnv
                .createTemporaryView(
                     "sensor" ,
                        stream,
                        $("id"),
                        $("temperature").as("temp"),
                        $("timestamp").as("ts"),
                        $("pt").proctime()//添加一个字段,作为处理时间
                );
        Table selectResult1 = TEnv.sqlQuery(
                "SELECT id, AVG(temp), TUMBLE_START(pt, INTERVAL '5' SECOND) FROM sensor GROUP BY id, TUMBLE(pt, INTERVAL '5' SECOND)"
        );
        //获取结果
        TEnv.toRetractStream(selectResult1,Row.class).print();

        env.execute();
    }
}
